package day08

import java.util.Properties
import java.util.concurrent.{ExecutorService, Executors}

import kafka.consumer.{Consumer, ConsumerConfig, ConsumerConnector, KafkaStream}
import kafka.message.MessageAndMetadata

import scala.collection.mutable

class ConsumerDemo(val consumer:String, val stream: KafkaStream[Array[Byte], Array[Byte]]) extends Runnable{
  override def run() = {
    val it = stream.iterator()
    while (it.hasNext()) {
      //取出数据
      val data: MessageAndMetadata[Array[Byte], Array[Byte]] = it.next()
      val topic: String = data.topic
      val offset: Long = data.offset
      val partition: Int = data.partition
      val msgByte: Array[Byte] = data.message()
      val msg = new String(msgByte)
      println(s"consumer: $consumer, topic: $topic, offset: $offset, partition: $partition, msg: $msg")
    }
  }
}

// 1,创建Consumer用Consumer.create方法 -> 需要参数 ConsumerConfig -> 需要 Properties，该类put方法添加所需配置如：group.id=，zookeeper.connect=，auto.offset.reset=，
// 2.获取流stream：consumer.createMessageStreams(定义的存储多个topic的map，可以设置为topics = new mutable.HashMap[String, Int]()该变量)
// 3.添加topic到topics里：topics.put(topic, 1)
// 4.用stream流来来拿某topic的数据：streams.get(topic)
// 5.创建线程池：Executors.newFixedThreadPool(3) 。
// 6.循环遍历拿到的数据总数topicStreams，然后用线程池.execute()方法来执行线程。里面的参数应该是一个实现了Runnable的本类如：threadPool.execute(new ConsumerDemo(s"consumer: $i", 拿到的数据读取每一条用.get(i)))
//
object ConsumerDemo {
  def main(args: Array[String]): Unit = {
    // 1.定义获取某topic
    val topic = "test1"
    // 2.定义map,用于存储多个topic的信息，key=topic,value=获取topic数据的线程数（相当于consumer的数量）
    val topics = new mutable.HashMap[String, Int]()
    // topic map用put方法添加k,v。
    topics.put(topic, 1)//如果该topic在kafka里只有一个partition的话，这里value如果指定为2,那么这里必定有一个是空着不干活的。
    // 配置propertirs
    val props = new Properties()
    // 定义group。属性在consumer.properties配置文件里的group.id=
    props.put("group.id","group01")
    // 指定zk列表:在consumer.properties配置文件里zookeeper.connect=master:2181,slave1:2181,slave2:2181
    props.put("zookeeper.connect","master:2181,slave1:2181,slave2:2181")
    // 指定offser:  在consumer.properties配置文件里auto.offset.reset=smallest
    // 如果zookeeper没有offset值或offset值超出范围。那么就给个初始的offset。
    props.put("auto.offset.reset","smallest")
    // 3.创建配置类，封装ConsumerConfig配置信息，传入上面properties对象
    val config = new ConsumerConfig(props)
    // 4.创建Consumer对象，调用Consumer.create(配置类)。。注意Consumer是Object类，直接就可以调用。
    val consumer: ConsumerConnector = Consumer.create(config)
    // 5.获取数据流：首先需要创建一个消息流MessageStream。map中，key=topic,value=topic对应的数据
    // Consumer实例对象.createMessageStreams(topic名)  得到一个Collect.Map[String, List[KafkaStream[array,array]]]
    val streams: collection.Map[String, List[KafkaStream[Array[Byte], Array[Byte]]]] = consumer.createMessageStreams(topics)
    // 把topic中的数据拿出来，将上面得到的流用..get(topic名)方法，返回值中的KafkaStream中,key=offset, value=data
    val topicStreams: Option[List[KafkaStream[Array[Byte], Array[Byte]]]] = streams.get(topic)
    // 6.创建固定线程池:来不断拿数据,,注意Executors.newCachedThreadPool(ThreadFactory)是创建可以复用的线程池
    val threadPool: ExecutorService = Executors.newFixedThreadPool(3)
    for (i <- 0 until topicStreams.size) {
      //线程池.execute()里面的参数应该是一个实现了Runnable的类
      threadPool.execute(new ConsumerDemo(s"consumer: $i", topicStreams.get(i)))
    }
  }
}
